package com.tiantang.hbase;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.stream.Collectors;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.NamespaceDescriptor;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Admin;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.ConnectionFactory;
import org.apache.hadoop.hbase.client.Delete;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.client.Table;
import org.apache.hadoop.hbase.filter.CompareFilter;
import org.apache.hadoop.hbase.filter.RowFilter;
import org.apache.hadoop.hbase.filter.SubstringComparator;
import org.apache.hadoop.hbase.util.Bytes;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * 
 * @description:功能描述 包含发布微博、关注、取关、浏览微博功能
 * @author: liujinkun
 * @date 2018年5月22日
 * @version:1.0
 */
public class WeiBo {

	private static Logger logger = LoggerFactory.getLogger(WeiBo.class);

	/**
	 * 命名空间
	 */
	private static String NAMESPACE = "ns_weibo";

	/**
	 * 微博内容表
	 */
	private static String CONTENT_TABLE = "ns_weibo:content";

	/**
	 * 用户关系表
	 */
	private static String USER_RELATION_TABLE = "ns_weibo:user_relation";

	/**
	 * 用户可查看的微博内容表（收件箱表）
	 */
	private static String INBOX_TABLE = "ns_weibo:inbox";

	private static Configuration conf = null;

	private static Connection connection = null;

	private static Admin admin = null;

	public WeiBo() {
		super();
		try {
			initConf();
			initConnection();
			initAdmin();
		} catch (IOException e) {
			logger.info("创建连接失败");
			e.printStackTrace();
		}
	}

	/**
	 * 
	 * @Title: initConf  
	 * @description:功能描述 初始化配置
	 * @return   
	 * @author:liujinkun
	 * @date 2018年5月22日
	 */
	public Configuration initConf() {
		if (conf == null) {
			conf = HBaseConfiguration.create();
			return conf;
		}
		return conf;
	}

	/**
	 * 
	 * @Title: initConnection  
	 * @description:功能描述 初始化连接
	 * @return
	 * @throws IOException   
	 * @author:liujinkun
	 * @date 2018年5月22日
	 */
	public Connection initConnection() throws IOException {
		if (connection == null) {
			connection=ConnectionFactory.createConnection(conf);
			return connection;
		}
		return connection;
	}

	/**
	 * 
	 * @Title: initAdmin  
	 * @description:功能描述 初始化HBase的表管理对象
	 * @return
	 * @throws IOException   
	 * @author:liujinkun
	 * @date 2018年5月22日
	 */
	public Admin initAdmin() throws IOException {
		if (admin == null) {
			admin=connection.getAdmin();
			return admin;
		}
		return admin;
	}

	/**
	 * 
	 * @Title: init  
	 * @description:功能描述 初始化命名空间和表
	 * @return   true：初始化成功；false：初始化失败
	 * @author:liujinkun
	 * @throws IOException 
	 * @date 2018年5月22日
	 */
	public boolean init() throws IOException {
//		this.initNameSpace();
//		this.initContentTable();
		this.initUserRelationTable();
//		this.initInboxTable();
		return true;
	}

	public void initNameSpace() throws IOException {
		NamespaceDescriptor namespaceDescriptor = NamespaceDescriptor.create(NAMESPACE)
				.addConfiguration("author", "TianTang").build();
		admin.createNamespace(namespaceDescriptor);
	}

	/**
	 * 
	 * @Title: initContentTable  
	 * @description:功能描述 创建微博内容表
	 * @throws IOException   
	 * @author:liujinkun
	 * @date 2018年5月22日
	 */
	public void initContentTable() throws IOException {
		// 创建表描述器
		HTableDescriptor contentTableDescriptor = new HTableDescriptor(TableName.valueOf(CONTENT_TABLE));
		// 创建列族描述器 该表只需要一个info列族
		HColumnDescriptor family = new HColumnDescriptor(Bytes.toBytes("info"));
		// 设置版本确界
		family.setMaxVersions(1);
		family.setMinVersions(1);
		// 设置缓存块
		family.setBlockCacheEnabled(true);
		// 设置缓存块的大小 2M
		family.setBlocksize(2 * 1024 * 1024);
		// 设置压缩式方式
//		family.setCompressionType(Algorithm.SNAPPY);
		// 添加列族
		contentTableDescriptor.addFamily(family);
		// 创建表
		admin.createTable(contentTableDescriptor);
	}

	/**
	 * 
	 * @Title: initUserRelationTable  
	 * @description:功能描述 创建用户关系表 
	 * @throws IOException   
	 * @author:liujinkun
	 * @date 2018年5月22日
	 */
	public void initUserRelationTable() throws IOException {
		HTableDescriptor userRelationTableDescriptor = new HTableDescriptor(TableName.valueOf(USER_RELATION_TABLE));

		// 创建关注列族
		HColumnDescriptor attendsFamily = new HColumnDescriptor(Bytes.toBytes("attends"));
		// 设置版本为1
		attendsFamily.setMinVersions(1);
		attendsFamily.setMaxVersions(1);
		// 设置缓存 大小为2M
		attendsFamily.setBlockCacheEnabled(true);
		attendsFamily.setBlocksize(2 * 1024 * 1024);
		// 设置压缩算法
//		attendsFamily.setCompressionType(Algorithm.SNAPPY);

		// 创建粉丝列族
		HColumnDescriptor fansFamily = new HColumnDescriptor(Bytes.toBytes("fans"));
		// 设置版本为1
		fansFamily.setMinVersions(1);
		fansFamily.setMaxVersions(1);
		// 设置缓存 大小为2M
		fansFamily.setBlockCacheEnabled(true);
		fansFamily.setBlocksize(2 * 1024 * 1024);
		// 设置压缩算法
//		fansFamily.setCompressionType(Algorithm.SNAPPY);

		// 添加关注列族
		userRelationTableDescriptor.addFamily(attendsFamily);
		// 添加粉丝列族
		userRelationTableDescriptor.addFamily(fansFamily);
		// 创建用户关系表
		admin.createTable(userRelationTableDescriptor);
	}

	/**
	 * 
	 * @Title: initInboxTable  
	 * @description:功能描述 创建收件箱表
	 * @throws IOException   
	 * @author:liujinkun
	 * @date 2018年5月22日
	 */
	public void initInboxTable() throws IOException {
		HTableDescriptor inboxTableDescriptor = new HTableDescriptor(TableName.valueOf(INBOX_TABLE));
		// 创建列族 只需要info列族
		HColumnDescriptor family = new HColumnDescriptor(Bytes.toBytes("info"));
		// 设置版本为1000
		family.setMinVersions(1);
		family.setMaxVersions(1000);
		// 设置缓存 2M
		family.setBlockCacheEnabled(true);
		family.setBlocksize(2 * 1024 * 1024);
		// 设置压缩算法
//		family.setCompressionType(Algorithm.SNAPPY);
		// 添加列族
		inboxTableDescriptor.addFamily(family);
		// 创建收件箱表
		admin.createTable(inboxTableDescriptor);
	}

	/**
	 * 
	 * @Title: publish  
	 * @description:功能描述 发布微博
	 * @param uid 发布者id
	 * @param content 微博内容
	 * @return  true：发布成功；false：发布失败 
	 * @author:liujinkun
	 * @throws IOException 
	 * @date 2018年5月22日
	 */
	public boolean publish(String uid, String content) throws IOException {
		Table contentTable = connection.getTable(TableName.valueOf(CONTENT_TABLE));
		// 1.向微博表添加一条微博
		long ts = System.currentTimeMillis();
		// 微博表的rowKey 为uid+'_'+时间戳
		String rowKey = uid + "_" + ts;
		Put contentPut = new Put(Bytes.toBytes(uid + "_" + ts));
		contentPut.addColumn(Bytes.toBytes("info"), Bytes.toBytes("content"), Bytes.toBytes(content));
		contentTable.put(contentPut);

		// 2.查询出该用户的粉丝，然后向每个粉丝的收件箱中添加一条微博
		Table userRelationTable = connection.getTable(TableName.valueOf(USER_RELATION_TABLE));
		// 查询粉丝
		Get get = new Get(Bytes.toBytes(uid));
		get.addFamily(Bytes.toBytes("fans"));
		Result fansResult = userRelationTable.get(get);
		List<Cell> cells = fansResult.listCells();
		// 没有查询到数据,直接结束微博的发布
		if(cells == null)
			return true;
		// 获取所有的粉丝id
		List<byte[]> fansUidList = cells.stream().map(e -> CellUtil.cloneValue(e)).collect(Collectors.toList());
		// 没有粉丝,直接结束微博的发布
		if (fansUidList.size() <= 0)
			return true;

		Table inboxTable = connection.getTable(TableName.valueOf(INBOX_TABLE));
		List<Put> inboxPuts = new ArrayList<Put>();
		// 遍历所有的粉丝id，然后给每个粉丝的收件箱添加一条微博
		for (byte[] fansUid : fansUidList) {
			Put put = new Put(fansUid);
			put.addColumn(Bytes.toBytes("info"), Bytes.toBytes(uid), ts, Bytes.toBytes(rowKey));
			inboxPuts.add(put);
		}
		inboxTable.put(inboxPuts);
		
		inboxTable.close();
		userRelationTable.close();
		contentTable.close();
		return true;
	}

	/**
	 * 
	 * @Title: attend  
	 * @description:功能描述 关注用户 
	 * @param uid 关注者id
	 * @param attendedId 被关注者id
	 * @return   true:关注成功，false：关注失败
	 * @author:liujinkun
	 * @throws IOException 
	 * @date 2018年5月22日
	 */
	public boolean attend(String uid, String attendedId) throws IOException {
		// 1.向关注者的关注列族中添加一列值；
		Table userRelationTable = connection.getTable(TableName.valueOf(USER_RELATION_TABLE));
		Put attendPut = new Put(Bytes.toBytes(uid));
		attendPut.addColumn(Bytes.toBytes("attends"), Bytes.toBytes(attendedId), Bytes.toBytes(attendedId));
		userRelationTable.put(attendPut);

		// 2.向被关注者的粉丝列族中添加一列值；
		Put fansPut = new Put(Bytes.toBytes(attendedId));
		fansPut.addColumn(Bytes.toBytes("fans"), Bytes.toBytes(uid), Bytes.toBytes(uid));
		userRelationTable.put(fansPut);

		// 3.向关注者的收件箱中添加被关注者的微博
		// 查询被关注对象的所有微博的rowKey
		Table contentTable = connection.getTable(TableName.valueOf(CONTENT_TABLE));
		Scan scan = new Scan();
		RowFilter filter = new RowFilter(CompareFilter.CompareOp.EQUAL, new SubstringComparator(attendedId));
		scan.setFilter(filter);
		scan.addFamily(Bytes.toBytes("info"));
		ResultScanner scanerResult = contentTable.getScanner(scan);
		List<byte[]> contentRowkeys = new ArrayList<byte[]>();
		for (Result result : scanerResult) {
			logger.info("微博的rowKey={}", Bytes.toString(result.getRow()));
			contentRowkeys.add(result.getRow());
		}
		// 向关注者的收件箱中添加微博
		Table inboxTable = connection.getTable(TableName.valueOf(INBOX_TABLE));
		Put inboxPut = new Put(Bytes.toBytes(uid));
		for (byte[] contentRowKey : contentRowkeys) {
			// 微博的rowKey 字符串形式
			String rowKey = Bytes.toString(contentRowKey);
			// 时间戳
			String ts = rowKey.split("_")[1];
			// 向收件箱添加微博，列名为被关注着的id，即微博的发布者
			inboxPut.addColumn(Bytes.toBytes("info"), Bytes.toBytes(attendedId), Long.parseLong(ts), contentRowKey);
		}
		inboxTable.put(inboxPut);
		
		inboxTable.close();
		contentTable.close();
		userRelationTable.close();
		return true;
	}

	/**
	 * 
	 * @Title: unfollow  
	 * @description:功能描述 取消关注
	 * @param uid 关注者id
	 * @param attendedId	被关注者id
	 * @return   true:取关成功；false：取关失败
	 * @author:liujinkun
	 * @throws IOException 
	 * @date 2018年5月22日
	 */
	public boolean unfollow(String uid, String attendedId) throws IOException {
		// 1.将关注者的attends列族中删除一列值
		Table userRelationTable = connection.getTable(TableName.valueOf(USER_RELATION_TABLE));
		Delete attendsDelete = new Delete(Bytes.toBytes(uid));
		attendsDelete.addColumn(Bytes.toBytes("attends"), Bytes.toBytes(attendedId));
		userRelationTable.delete(attendsDelete);
		
		// 2.将被关注者的fans列族中删除一列值
		Delete fansDelete = new Delete(Bytes.toBytes(attendedId));
		fansDelete.addColumn(Bytes.toBytes("fans"), Bytes.toBytes(uid));
		userRelationTable.delete(fansDelete);
		
		// 3.将关注者的收件箱中相应微博删除
		Table inboxTable = connection.getTable(TableName.valueOf(INBOX_TABLE));
		Delete inboxDelete = new Delete(Bytes.toBytes(uid));
		// 如果使用 addColumns 函数，则会删除该行中的指定列中所有的数据。
		// 如果使用 addColumn() 函数则会删除改行中指定列的最大本版本的KeyValue数据。
		inboxDelete.addColumns(Bytes.toBytes("info"), Bytes.toBytes(attendedId));
		inboxTable.delete(inboxDelete);
		
		inboxTable.close();
		userRelationTable.close();
		return true;
	}

	/**
	 * 
	 * @Title: browse  
	 * @description:功能描述 浏览微博
	 * @param uid   浏览者id
	 * @author:liujinkun
	 * @throws IOException 
	 * @date 2018年5月22日
	 */
	public void browse(String uid) throws IOException {
		// 1.查询查看者的收件箱
		Table inboxTable = connection.getTable(TableName.valueOf(INBOX_TABLE));
		Get inboxGet = new Get(Bytes.toBytes(uid));
		inboxGet.addFamily(Bytes.toBytes("info"));
		// 只浏览最近五条微博
		inboxGet.setMaxVersions(5);
		Result result = inboxTable.get(inboxGet);
		List<Cell> cells = result.listCells();
		if(cells == null){
			logger.info("您目前没有微博可以浏览，快去关注大V吧");
			return ;
		}
			
		List<byte[]> rowKeys=new ArrayList<byte[]>();
		for (Cell cell : cells) {
			// 获取列值(微博的rowKey)
			byte[] rowKey=CellUtil.cloneValue(cell);
			rowKeys.add(rowKey);
		}
		
		// 根据微博的 rowKey 查询微博
		List<Get> contentGets=new ArrayList<>();
		// 构建Get 对象
		for (byte[] rowKey : rowKeys) {
			Get contentGet =new Get(rowKey);
			contentGet.addFamily(Bytes.toBytes("info"));
			contentGets.add(contentGet);
		}
		Table contentTable=connection.getTable(TableName.valueOf(CONTENT_TABLE));
		Result[] contentResults = contentTable.get(contentGets);
		List<Message> messages = new ArrayList<Message>();
		// 遍历结果集（一条微博对应一条结果）
		for (Result r : contentResults) {
			// 微博的 rowKey 字符串
			String rowKey=Bytes.toString(r.getRow());
			// 微博发布者uid
			String contentUid=rowKey.split("_")[0];
			// 微博发布时间
			String ts=rowKey.split("_")[1];
			
			Cell[] contentCells=r.rawCells();
			for (Cell cell : contentCells) {
				Message message=new Message();
				message.setUid(contentUid);
				message.setDate(DateUtil.dateToStr(Long.parseLong(ts), "yyyy-MM-dd HH:mm:ss"));
				message.setContent(Bytes.toString(CellUtil.cloneValue(cell)));
				messages.add(message);
			}
		}
		logger.info("可浏览的微博条数为：number={}",messages.size());
		logger.info(messages.toString());
		inboxTable.close();
	}
}
